"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : Logical_Replication
Case Name   : 在白名单里的系统管理员创建、开启和删除逻辑复制槽失败，合理报错
Create At   : 2023/9/23
Owner       : @xiaqi14
Description :
    1.修改参数
    2.创建sysadmin用户，配置用户白名单
    3.创建逻辑复制槽
    4.开启逻辑复制槽
    5.删除逻辑复制槽
    6.删除用户
    7.恢复参数，恢复pg_hba.conf文件
Expect      :
    1.修改参数成功
    2.创建用户成功，配置白名单成功
    3.创建失败，合理报错
    4.开启失败，合理报错
    5.删除失败，合理报错
    6.删除成功
    7.恢复成功
History     :
"""
import os
import unittest
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant


class LogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.sh_primary = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.user_name = 'u_logical_replication_0064'
        self.slot_name = 'slot_logical_replication_0064'
        self.config_wal_level = 'wal_level'
        self.wal_level_default = self.common.show_param(self.config_wal_level)
        self.config_duty = 'enableSeparationOfDuty'
        self.duty_default = self.common.show_param(self.config_duty)

    def test_logical_replication(self):
        text = "-----step1:修改参数;expect:修改参数成功-----"
        self.logger.info(text)
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primary.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.common.show_param(self.config_wal_level)
        self.assertEquals(show_msg, 'logical', '执行失败' + text)

        if 'on' != self.duty_default:
            result = self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_duty}=on')
            self.assertTrue(result, '执行失败' + text)
            restart_res = self.sh_primary.restart_db_cluster()
            self.assertTrue(restart_res, '执行失败' + text)
        show_msg = self.common.show_param(self.config_duty)
        self.assertEquals(show_msg, 'on', '执行失败' + text)

        text = "-----step2:创建sysadmin用户，配置用户白名单，" \
               "expect:创建用户成功，配置白名单成功-----"
        self.logger.info(text)
        sql_cmd = f'''drop user if exists {self.user_name};
            create user {self.user_name} sysadmin 
            password '{macro.COMMON_PASSWD}';'''
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG and
                      self.constant.CREATE_ROLE_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)
        param_1 = f'local replication {self.user_name}              sha256'
        param_2 = f'host  replication {self.user_name} 127.0.0.1/32 sha256'
        param_3 = f'host  replication {self.user_name} ::1/128      sha256'
        result_1 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step3:创建逻辑复制槽;expect:创建失败，合理报错-----"
        self.logger.info(text)
        create_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--create', 'postgres', self.slot_name, self.user_name, '-s 2 -W')
        self.logger.info(create_slot_msg)
        self.assertIn("FATAL:  must be replication role or a member of "
                      "the gs_role_replication role to use replication slots "
                      "when separation of privileges is used",
                      create_slot_msg, '执行失败:' + text)

        text = "-----step4:开启逻辑复制槽;expect:开启失败，合理报错-----"
        self.logger.info(text)
        start_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--start', 'postgres', self.slot_name, self.user_name,
            '-f - -o skip-empty-xacts=true -F 1 -s 2 -W')
        self.logger.info(start_slot_msg)
        self.assertIn("FATAL:  must be replication role or a member of "
                      "the gs_role_replication role to use replication slots "
                      "when separation of privileges is used",
                      start_slot_msg, '执行失败:' + text)

        text = "-----step5:删除逻辑复制槽;expect:删除失败，合理报错-----"
        self.logger.info(text)
        drop_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--drop', 'postgres', self.slot_name, self.user_name, '-s 2 -W')
        self.logger.info(drop_slot_msg)
        self.assertIn("FATAL:  must be replication role or a member of "
                      "the gs_role_replication role to use replication slots "
                      "when separation of privileges is used",
                      drop_slot_msg, '执行失败:' + text)

    def tearDown(self):
        text_1 = "-----step6:删除用户;expect:删除成功-----"
        self.logger.info(text_1)
        drop_cmd = f'''drop user {self.user_name};'''
        drop_msg = self.sh_primary.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)

        text_2 = "-----step7:恢复参数，恢复pg_hba.conf文件，expect:恢复成功-----"
        self.logger.info(text_2)
        param_1 = f'local replication {self.user_name} '
        param_2 = f'host  replication {self.user_name} 127.0.0.1/32 '
        param_3 = f'host  replication {self.user_name} ::1/128 '
        result_1 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)

        wal_level_current = self.common.show_param(self.config_wal_level)
        if wal_level_current != self.wal_level_default:
            self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}={self.wal_level_default}')
            restart_msg = self.sh_primary.restart_db_cluster()
            self.logger.info(restart_msg)
        show_wal_level = self.common.show_param(self.config_wal_level)

        current_duty = self.common.show_param(self.config_duty)
        if self.duty_default != current_duty:
            self.guc_duty_result = self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_duty}={self.duty_default}')
            self.restart_res = self.sh_primary.restart_db_cluster()
        show_duty_msg = self.common.show_param(self.config_duty)

        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_msg, '执行失败' + text_1)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text_2)
        self.assertEquals(show_wal_level, self.wal_level_default,
                          '执行失败' + text_2)
        self.assertEquals(show_duty_msg, self.duty_default, '执行失败' + text_2)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
